package com.atguigu.flink.datastreamapi.agg;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2023/4/4
 */
public class Demo1_Agg
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<WaterSensor> ds = env.socketTextStream("hadoop102", 8888)
                                                               .map(new WaterSensorMapFunction());


        /*
            无分组(keyBy)，不聚合。

            KeyedStream: 按照key进行分组的流，可以进行聚合
         */
        KeyedStream<WaterSensor, String> ds1 = ds.keyBy(WaterSensor::getId);

        ds1
            //.sum("vc")
            /*
                除了分组和聚合的字段，其他字段取当前组内第一条数据的值。
             */
            //.max("vc")
            //.min("vc")
            /*
               maxBy("vc")： 希望除了分组和聚合的字段，其他字段取当前组内最大值数据的值。出现同样大的，输出第一条最大的值。

               maxBy("vc",)： 希望除了分组和聚合的字段，其他字段取当前组内最大值数据的值。出现同样大的，输出最新的值。

             */
            .maxBy("vc",false)
           .print();




        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
